package idempotence;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/3/18 18:42
 * @Version: 1.0
 * @description:幂等性操作的参数配置
 */
public class KafkaProdecerDemo {

    public static void main(String[] args) {

        //创建配置对象，指定producer的配置
        Properties properties = new Properties();

        //配置与kafka集群连接的参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");

        //配置producer发布的recoder的key进行序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

        //配置producer发布的recoder的value进行序列化,StringSerializer序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        //配置幂等性的ack响应机制，all或者-1代表，消息发布到leader分区以及follower分区，然后触发ack响应机制
        properties.put(ProducerConfig.ACKS_CONFIG,"all");

        //设置重发的次数限定
        properties.put(ProducerConfig.RETRIES_CONFIG,3);

        //设置请求的超时间 单位毫秒
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);

       // 开启幂等性支持
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        //设置批量发送数据的上限值 1kb=1024bit(字节)
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);

        //批量发送的时间设置，时间设置优于数据量设置（发送时间达到，为达到数据量上线，也要进行发送数据）设置1000毫秒后进行发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1000);

        //创建producer对象
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(properties);



        for (int i = 20; i <30 ; i++) {
            ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("baizhi",i,"nihaowodeshijie");
            producer.send(record);
        }
        //发布消息


        //提交
        producer.flush();

       //释放资源
       producer.close();



    }




}
